package com.atguigu.upp.app;

import com.atguigu.upp.beans.TagInfo;
import com.atguigu.upp.service.CKService;
import com.atguigu.upp.service.MysqlService;
import com.atguigu.upp.util.UPPUtil;
import jodd.util.PropertiesUtil;
import lombok.extern.log4j.Log4j;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2023/3/25
 */
@Log4j
public class MergeWideTableApp
{
    public static void main(String[] args) throws IOException {

        //接收画像平台传入的参数
        String taskId = args[0];
        String doDate = args[1];

        /*String taskId = "3";
        String doDate = "2022-06-09";*/

        SqlSessionFactory mysqlSSF = UPPUtil.createSqlSessionFactory("mysqlconfig.xml");
        SqlSessionFactory ckSSF = UPPUtil.createSqlSessionFactory("ckconfig.xml");
        MysqlService mysqlService = new MysqlService(mysqlSSF.openSession(true));
        CKService ckService = new CKService(ckSSF.openSession(true));
        SparkSession sparkSession = UPPUtil.getSparkSession("MergeWideTableApp");

        List<TagInfo> tags = mysqlService.queryTags();

        //生成pivotsql
        String pivotSql = generatePivotSql(tags, doDate);

        //查询pivotsql，写入ck
        wrietToCk(pivotSql, ckService, doDate, sparkSession, tags);


    }

    private static void wrietToCk(String pivotSql, CKService ckService, String doDate, SparkSession sparkSession, List<TagInfo> tags) {

        //定义宽表的名字格式   前缀+日期  每天一张宽表
        String name = UPPUtil.getValue("upwideprefix") + doDate.replace("-", "_");

        //写入之前，先删表。保证可以重复执行
        ckService.dropWideTable(name);

        //建表
        String col = tags.stream().map(tag -> tag.getTagCode().toLowerCase() + " String ").collect(Collectors.joining(","));
        //列的名字就是标签的名字小写，类的类型这里使用一个通用的类型 String
        ckService.createWideTable(name, col);

        //查询pivotsql
        Dataset<Row> data = sparkSession.sql(pivotSql);

        //写入ck 使用SparkSql提供的api写入
        Properties properties = new Properties();
        //存放其他的参数，例如用户名，密码等
        data.write()
            //默认是 ErrorIfExists，特点是自动去建表，表已经存在就报错
            .mode(SaveMode.Append)
            .option("driver", UPPUtil.getValue("ck.jdbc.driver.name"))
            .option("batchsize", 500)
            .option("isolationLevel", "NONE")   //事务关闭
            .option("numPartitions", "4") // 设置并发
            .jdbc(UPPUtil.getValue("ck.jdbc.url"), name, properties);

    }

    /*
select
    *
from (
         select uid, `tagValue`, 'tag_population_attribute_nature_gender' tagCode
         from upp221109.tag_population_attribute_nature_gender
         where dt = '2022-06-09'
         union all
         select uid, `tagValue`, 'tag_population_attribute_nature_period' tagCode
         from upp221109.tag_population_attribute_nature_period
         where dt = '2022-06-09'
         union all
         select uid, `tagValue`, 'tag_consumer_behavior_order_amount7d' tagCode
         from upp221109.tag_consumer_behavior_order_amount7d
         where dt = '2022-06-09'
     ) t
    pivot (
       min(tagValue)
      for tagCode in( 'tag_population_attribute_nature_gender' ,
                     'tag_population_attribute_nature_period',
                     'tag_consumer_behavior_order_amount7d')
    );
     */
    private static String generatePivotSql(List<TagInfo> tags, String doDate) {

        String template = " select * from ( %s )t " +
            " pivot ( min( tagValue ) for tagCode in ( %s ) )  ";


        String singleTagSql = "   select uid, `tagValue`, '%s' tagCode  from %s.%s where dt = '%s' ";

        //查询库名
        String db = UPPUtil.getValue("updbname");

        //根据今天要计算的标签，填充singleTagSql
        String tmpTableSql = tags.stream()
                                 .map(tag -> String.format(singleTagSql, tag.getTagCode().toLowerCase(), db, tag.getTagCode().toLowerCase(), doDate))
                                 .collect(Collectors.joining(" union all  "));

        //生成旋转列值 标签名
        String tagNameSql = tags.stream()
                                .map(tag -> "'" + tag.getTagCode().toLowerCase() + "'")
                                .collect(Collectors.joining(","));

        //格式化
        String pivotSql = String.format(template, tmpTableSql, tagNameSql);

        log.warn(pivotSql);

        return pivotSql;
    }
}
